import com.atguigu.realtime.app.BaseAppV1;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/7/19 14:44
 */
public class Join_2 extends BaseAppV1 {
    public static void main(String[] args) {
        new Join_2().init(10000,2,"abc", "ab");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        //stream.print();
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
        /*tEnv.executeSql("create table ab(id string, name string, age int)with(" +
                            "'connector' = 'kafka', " +
                            "'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "'properties.group.id' = 'abcccc', " +
                            "'topic' = 'ab', " +
                            "'format' = 'json'" +
                            ")");*/
    
        tEnv.executeSql("create table ab(id string, name string, age int, primary key(id) not enforced) with(" +
                            "'connector' = 'upsert-kafka', " +
                            "'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "'topic' = 'ab', " +
                            "'key.format' = 'json', " +
                            "'value.format' = 'json' " +
                            ")");
        
        tEnv.sqlQuery("select * from ab").execute().print();
        
    }
}
